package cn.gwm.flink.streaming.function.source;
import cn.gwm.flink.streaming.beans.termalRunaway.BmsTrdStrategys;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.task.ThermalWarningEvDwmTask;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.db.Db;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.json.JSONUtil;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 热失控策略数据流
 *
 * @author lxy
 */
public class TermalBroadcastSourceFunction extends RichParallelSourceFunction<Map<String, List>> {
    private boolean flag = true;
    private static final Logger logger = LoggerFactory.getLogger(TermalBroadcastSourceFunction.class);
    private String env;
    public TermalBroadcastSourceFunction(String env){
        this.env = env;
    }
    @Override
    public void run(SourceContext<Map<String, List>> sourceContext) {
        try {
            Map<String, List> map = new ConcurrentHashMap(0);
            List<BmsTrdStrategys> bmsTrdStrategies = new ArrayList<>(0);
            while (flag) {
                synchronized (this){
                   if("test".equals(env)){
                        bmsTrdStrategies = Db.use(DSFactory.get("group_aliyun_db1")).find(BmsTrdStrategys.getEntityWhere(), BmsTrdStrategys.class);
                    }else{
                        bmsTrdStrategies = Db.use(DSFactory.get("group_db1")).find(BmsTrdStrategys.getEntityWhere(), BmsTrdStrategys.class);
                    }
                    map.put("BmsTrdStrategy",bmsTrdStrategies);
                    sourceContext.collect(map);
                    map.clear();
                    bmsTrdStrategies.clear();
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            logger.error("广播流mysql连接异常e.getStackTrace：{}",e.getStackTrace());
            logger.error("广播流mysql连接异常e.getMessage：{}",e.getMessage());
            logger.error("广播流mysql连接异常e.getCase：{}",e.getCause());
        }

    }

    @Override
    public void cancel() {
        flag = false;
    }
}
